package org.example.producer;

import org.example.topic.Topics;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.UUID;

@RestController
public class KafkaProducerController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;


    @GetMapping("/send")
    public HashMap send(@RequestParam("msg") String msg) {
        System.out.println("发送msg=" + msg);
        ListenableFuture<SendResult<String, String>> resultListenableFuture = kafkaTemplate.send(Topics.TEST, 0, UUID.randomUUID().toString(), msg);
        SendResult<String, String> sendResult = null;
        try {
            sendResult = resultListenableFuture.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("sendResult=" + sendResult);
        HashMap<Object, Object> hashMap = new HashMap<>();
        hashMap.put("data", "ok");
        return hashMap;
    }


}
